package com.example.actordemo3.pubSub;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.cluster.pubsub.DistributedPubSub;
import akka.cluster.pubsub.DistributedPubSubMediator;
import akka.event.Logging;
import akka.event.LoggingAdapter;

public class PubSubTwo extends AbstractActor {
    LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    ActorRef mediator;

    final  String topic = "allSubPub";
    @Override
    public void preStart() throws Exception {
         mediator = DistributedPubSub.get(getContext().system()).mediator();

         this.mediator.tell(new DistributedPubSubMediator.Subscribe(topic, getSelf()), ActorRef.noSender());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder() .match(String.class, msg -> log.info("Got: {}", msg))
                .match(DistributedPubSubMediator.SubscribeAck.class, msg -> log.info("subscribed"))
                .match(Integer.class,msg->{
                    this.mediator.tell(new DistributedPubSubMediator.Subscribe(topic, getSelf()), ActorRef.noSender());
                })
                .matchAny(mag->{
                    System.out.println("1111111111");

        }).build();
    }


    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("sys1");
        system.actorOf(Props.create(pubSubTets.Subscriber.class), "subscriber1");
    }
}

 class Publisher extends AbstractActor {

    // activate the extension
    ActorRef mediator = DistributedPubSub.get(getContext().system()).mediator();

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(
                        String.class,
                        in -> {
                            String out = in.toUpperCase();
                            mediator.tell(new DistributedPubSubMediator.Publish("content", out), getSelf());
                        })
                .build();
    }
}

